package com.visec.hdy.hadoop.storage;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.nio.charset.Charset;
import java.util.List;

import org.apache.log4j.Logger;

import com.visec.hdy.hadoop.standard.Analysis.Item;
import com.visec.hdy.hadoop.standard.FSB;
import com.visec.hdy.hadoop.standard.Storage;

/**
 * 默认数据存储实现类
 * 
 * @author hdy
 *
 * @param <K>
 * @param <V>
 */
public class LogStorageImpl<K, V> extends DefaultStorageImpl<K, V> implements Storage<K, V> {
	protected BufferedWriter writer;
	protected static final  String DEFAULT_CODE = "UTF-8";
	protected static final char DELIMITER = '=';
	private final Logger l = Logger.getLogger(LogStorageImpl.class);

	public LogStorageImpl(FSB fsb) {
		super(fsb);
	}

	protected String dataFormat(Item<K, V> item) {
		return item.getKey().toString() + DELIMITER + item.getObject().toString();
	}

	@Override
	public void store(Item<K, V> item) {
		try {
			if (writer == null || (!context.write())) {
				newConnect(item.getKey());
			}
			writer.write(dataFormat(item));
			writer.newLine();
		} catch (IOException e) {
			l.error("", e);
		}
	}

	public void newConnect(K k) {
		close();
		outputStream = fsb.create(context.newFile(k));
		writer = new BufferedWriter(new OutputStreamWriter(outputStream, Charset.forName(DEFAULT_CODE)));
	}

	public void connect(File file) {
		outputStream = fsb.append(file);
		writer = new BufferedWriter(new OutputStreamWriter(outputStream, Charset.forName(DEFAULT_CODE)));
	}

	@Override
	public void close() {
		if (writer != null)
			try {
				writer.close();
			} catch (IOException e) {
				l.error("", e);
			}
		super.close();
	}

	@Override
	public void open() {
		close();
		List<File> listFiles = context.sortFiles(fsb.listFiles(new File(context.getFolder())));
		if (!listFiles.isEmpty()) {
			File file = listFiles.get(listFiles.size() - 1);
			try (InputStream inputStream = fsb.get(file);
					InputStreamReader isr = new InputStreamReader(inputStream);
					BufferedReader br = new BufferedReader(isr)) {
				context.skip(br.lines().count() - 1L);
			} catch (Exception e) {
				l.error("", e);
			}
			connect(file);
		}
	}

	@Override
	public void flush() {
		try {
			writer.flush();
		} catch (IOException e) {
			l.error("", e);
		}
	}

}
